package io.wen.bd.s8m1;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.LocalDate;

import static org.apache.flink.table.api.Expressions.$;

public class Task5_FlinkTable {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        final Table rawCustomers =
                tEnv.fromValues(
                        Row.of("Guillermo Smith",LocalDate.parse("1992-12-12"),"4081 Valley Road","08540","New Jersey","m",true,0,78,3),
                        Row.of("Valeria Mendoza",LocalDate.parse("1970-03-28"),"1239  Rainbow Road","90017","Los Angeles","f",true,9,39,0),
                        Row.of("Leann Holloway",LocalDate.parse("1989-05-21"),"2359 New Street","97401","Eugene",null,true,null,null,null),
                        Row.of("Brandy Sanders",LocalDate.parse("1956-05-26"),"4891 Walkers-Ridge-Way","73119","Oklahoma City","m",false,9,39,0),
                        Row.of("John Turner",LocalDate.parse("1982-10-02"),"2359 New Street","60605","Chicago","m",true,12,39,0),
                        Row.of("Ellen Ortega",LocalDate.parse("1985-06-18"),"2448 Rodney STreet","85023","Phoenix","f",true,0,78,3));

        final Table namedCustomers = rawCustomers.as("name","date_of_birth","street","zip_code","city","gender","has_newsletter");
        tEnv.createTemporaryView("customers", namedCustomers);
        final Table selected = tEnv.from("customers")
                .filter($("gender").isNotNull())
                .select($("name").upperCase(),
                        $("date_of_birth"));
        final DataStream<Tuple2<Boolean, Row>> result = tEnv.toRetractStream(selected, Row.class);
        result.print();

        env.execute();
    }
}
